package de.preissler.insertjoin;

import java.util.LinkedList;
import java.util.List;
import java.util.Map;

import me.prettyprint.hector.api.Keyspace;
import me.prettyprint.hector.api.beans.HColumn;
import me.prettyprint.hector.api.factory.HFactory;
import me.prettyprint.hector.api.mutation.Mutator;

public class InsertDataPreSaved implements InsertData
{

	private static final String JOIN = "Join_Quotes_Dax";
	private static final String VALUE = "value";

	public void insert(Map<String, Map<String, Quote>> quotes,
			Map<String, Map<String, DAXParam>> params, Keyspace keyspace,
			Mutator<String> mutator)
	{
		insertQuotesInBase(quotes, keyspace, mutator);
		insertDAXParamInBase(params, keyspace, mutator);
		insertJoinQuotesDAX(quotes, params, keyspace, mutator);
		mutator.execute();

	}
	
	private void insertJoinQuotesDAX(Map<String, Map<String, Quote>> quotes,
			Map<String, Map<String, DAXParam>> params, Keyspace keyspace,
			Mutator<String> mutator)
	{
		for(String date : params.keySet())
		{
			double divident = 0.0;
			double divisor = 0.0;
			for(String isin : params.get(date).keySet())
			{	
				if(!isin.equals(DAX_ISIN))
				{
					double qit = params.get(date).get(isin).qit;
					double qi0 = params.get(date).get(isin).qi0;
					double pit = quotes.get(date).get(isin).close;
					double ci = params.get(date).get(isin).ci;
					double ffi = params.get(date).get(isin).ffi;
					double pi0 = params.get(date).get(isin).pi0;
					
					divident = divident + (pit * qit * ffi * ci);
					divisor = divisor + (pi0 * qi0);
				}
			
			}
			//DAX zu ende berechnen und einfuegen
			List<HColumn<String, Double>> columnList = new LinkedList<HColumn<String, Double>>();
			double kt = params.get(date).get(DAX_ISIN).kt;
			double base = params.get(date).get(DAX_ISIN).base;
			double value = kt * (divident/divisor) * base;
			columnList.add(HFactory.createColumn(KT, kt, SS, DS));
			columnList.add(HFactory.createColumn(BASE, base, SS, DS));
			columnList.add(HFactory.createColumn(VALUE, value, SS, DS));
			mutator.addInsertion(date, JOIN, HFactory.createSuperColumn(DAX_ISIN, columnList, SS, SS, DS));
		}
	}

	private void insertQuotesInBase(Map<String, Map<String, Quote>> quotes, Keyspace keyspace, Mutator<String> mutator)
	{
		for(String isin : quotes.get(quotes.keySet().iterator().next()).keySet())
		{
			for(String date : quotes.keySet())
			{
				List<HColumn<String, Double>> columnList = new LinkedList<HColumn<String, Double>>();
				
				//Werte holen
				columnList.add(HFactory.createColumn(CHANGE, quotes.get(date).get(isin).change, SS, DS));
				columnList.add(HFactory.createColumn(CLOSE, quotes.get(date).get(isin).close, SS, DS));
				columnList.add(HFactory.createColumn(HIGH, quotes.get(date).get(isin).high, SS, DS));
				columnList.add(HFactory.createColumn(LOW, quotes.get(date).get(isin).low, SS, DS));
				columnList.add(HFactory.createColumn(OPEN, quotes.get(date).get(isin).open, SS, DS));
				columnList.add(HFactory.createColumn(VOLUME, quotes.get(date).get(isin).volume, SS, DS));
				
				mutator.addInsertion(isin, QUOTES, HFactory.createSuperColumn(date, columnList, SS, SS, DS));
			}
		}
	}
	
	private void insertDAXParamInBase(Map<String, Map<String, DAXParam>> params, Keyspace keyspace, Mutator<String> mutator)
	{
		for(String date : params.keySet())
		{
			for(String isin : params.get(date).keySet())
			{
				List<HColumn<String, Double>> columnList = new LinkedList<HColumn<String, Double>>();
				
				if(isin.equals(DAX_ISIN))
				{
					columnList.add(HFactory.createColumn(KT, params.get(date).get(isin).kt, SS, DS));
					columnList.add(HFactory.createColumn(BASE, params.get(date).get(isin).base, SS, DS));
				}
				else
				{
					columnList.add(HFactory.createColumn(QIT, params.get(date).get(isin).qit, SS, DS));
					columnList.add(HFactory.createColumn(QI0, params.get(date).get(isin).qi0, SS, DS));
					columnList.add(HFactory.createColumn(CI, params.get(date).get(isin).ci, SS, DS));
					columnList.add(HFactory.createColumn(FFI, params.get(date).get(isin).ffi, SS, DS));
					columnList.add(HFactory.createColumn(PI0, params.get(date).get(isin).pi0, SS, DS));
				}
				
				mutator.addInsertion(date, DAX, HFactory.createSuperColumn(isin, columnList, SS, SS, DS));
			}
		}
	}

	
	public void delete(Map<String, Map<String, Quote>> quotes,
			Map<String, Map<String, DAXParam>> params, Keyspace keyspace,
			Mutator<String> mutator)
	{
		deleteQuotesinBase(quotes, keyspace, mutator);
		deleteDAXParamsinBase(params, keyspace, mutator);
		deleteJoin(quotes, params, keyspace, mutator);
		mutator.execute();
		
		//Compaction in cassandra triggern
		JMX.clean(Main.HOST, Main.KEYSPACE_NAME, InsertData.QUOTES);
		JMX.clean(Main.HOST, Main.KEYSPACE_NAME, InsertData.DAX);
		JMX.clean(Main.HOST, Main.KEYSPACE_NAME, InsertDataPreSaved.JOIN);
	}
	
	private void deleteJoin(Map<String, Map<String, Quote>> quotes,
			Map<String, Map<String, DAXParam>> params, Keyspace keyspace,
			Mutator<String> mutator)
	{
		for(String date : params.keySet())
		{
			for(String isin : params.get(date).keySet())
			{
				mutator.addDeletion(date, JOIN, isin, SS);
			}
		}
	}

	private void deleteQuotesinBase(Map<String, Map<String, Quote>> quotes,
			Keyspace keyspace, Mutator<String> mutator)
	{
		for(String isin : quotes.get(quotes.keySet().iterator().next()).keySet())
		{
			for(String date : quotes.keySet())
			{
				mutator.addDeletion(isin, QUOTES, date, SS);
			}
		}
	}
	
	private void deleteDAXParamsinBase(Map<String, Map<String, DAXParam>> params, Keyspace keyspace, Mutator<String> mutator)
	{
		for(String date : params.keySet())
		{
			for(String isin : params.get(date).keySet())
			{
				mutator.addDeletion(date, DAX, isin, SS);
			}
		}
	}
}
